package com.atguigu.flink.sqlfunction;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 *  标量函数： 普通的UDF，一进一出
 */
public class Demo2_ScalarFunction
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = " create table t1 ( id string,ts bigint, vc int  )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);
        Table table = tableEnvironment.from("t1");

        //创建自己定义的函数对象
        MyUpper myUpper = new MyUpper();

        //方式一：在TableAPI中使用。可以起，可以不起
        //table.select($("id"), Expressions.call(myUpper,$("id"))).execute().print();

        //为函数起名字
        tableEnvironment.createTemporaryFunction("myUpper",myUpper);
        //table.select($("id"), Expressions.call("myUpper",$("id"))).execute().print();

        //方式二：在sql中使用。 必须为函数起名字
        tableEnvironment.sqlQuery("select id,myUpper(id) from t1").execute().print();

    }

    /*
        把字符串转大写
        想要实现自定义标量函数，你需要扩展 org.apache.flink.table.functions 里面的 ScalarFunction 并且实现一个或者多个求值方法。
        标量函数的行为取决于你写的求值方法。求值方法必须是 public 的，而且名字必须是 eval。
     */
    public static class MyUpper extends ScalarFunction{

        //不能是void
        public String eval(String str){
            if (StringUtils.isBlank(str)){
                return "NULL";
            }else{
                return str.toUpperCase();
            }
        }
    }

}
